package com.tbdp.lsmp.processor;

import com.google.common.base.Joiner;

import com.tbdp.lsmp.common.Configuration;
import com.tbdp.lsmp.common.Constants;
import com.tbdp.lsmp.util.DateUtils;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import tbdp.tes.es.service.EsSqlClient;
import tbdp.tes.es.service.UserCache;
import tbdp.tes.expression.Expression;
import tbdp.tes.expression.ExpressionUtil;
import tbdp.tes.load.ESModelQueryLoad;
import tbdp.tes.model.ModConfig;
import tbdp.tes.model.Model;
import tbdp.tes.model.Rule;
import tbdp.tes.model.StrategyConfig;
import tbdp.tes.model.UserVari;
import tbdp.tes.model.UserVariConfig;
import tbdp.tes.model.Vari;

/**
 * 查询es数据
 * 1.首先需要加载mysql数据，解析出需要加载的es column列表
 * 2.load 全量的es数据，在内存里面把规则过滤下
 * 3.把服务规则的结果落到kafka 里面
 *
 * @author xbkaishui
 * @version $Id: ESQueryProcessor.java, v 0.1 2017-11-06 下午8:59 xbkaishui Exp $$
 */
public class ESQueryProcessor {

    private static Logger logger = LoggerFactory.getLogger(ESQueryProcessor.class);

    private KafkaProcessor kafkaProcessor;

    private EsSqlClient esSqlClient;

    private ESModelQueryLoad queryLoad = new ESModelQueryLoad();

    private Expression expression;



    public void init() {
        if (kafkaProcessor == null) {
            logger.info("init kafka processor");
            kafkaProcessor = new KafkaProcessor();
        }
        if (esSqlClient == null) {
            logger.info("init es sql client");
            esSqlClient = EsSqlClient.instance();
        }

        if (expression == null) {
            logger.info("init expression");
            expression = ExpressionUtil.getExpression();
        }
        kafkaProcessor.init();
    }

    public void process() {
        try {

            List<Model> models = queryLoad.batchQueryOnlineModel();
            if (models == null) {
                logger.warn("models is empty!");
                return;
            }
            //访问es的列列表
            Set<String> needFields = new HashSet<String>();

            getAllColumns(models, needFields);

            logger.info("all needFields {} ", needFields);

            StringBuffer sqlBuf = new StringBuffer();
            String esIndex = Configuration.getInstance().getProperty(Constants.ES_INDEX);
            String esLimit = Configuration.getInstance().getProperty(Constants.ES_LIMIT);
            sqlBuf.append("select ").append(Joiner.on(" , ").join(needFields));
            sqlBuf.append(" from ").append(esIndex);
            sqlBuf.append(" limit  " + Integer.valueOf(esLimit));

            logger.warn("es query sql {} ", sqlBuf.toString());
            //查询es 客户端
            List<Map<String, String>> dataResults = esSqlClient.query(sqlBuf.toString(), needFields.toArray(new String[]{}));

            // expression 解析表达式
            for (Model model : models) {
                logger.info(" handle model " +model.getModConfig().getModId());
                try {
                    for (Map<String, String> dataResult : dataResults) {
                        boolean allPasses = true;
                        List<Rule> rules = model.getRules();
                        for (Rule rule : rules) {
                            allPasses = allPasses && evalRule(rule, dataResult);
                            if (!allPasses) {
                                break;
                            }
                        }
                        if (allPasses) {
                            sendOut(model, dataResult);
                        }
                    }
                } catch (Throwable e) {
                    logger.error("es process model error " + model.getModConfig().getModId(), e);
                }

            }

        } catch (Throwable e) {
            logger.error("es process error ", e);
        }

    }

    /**
     * 写出匹配到的股票结果
     */
    private void sendOut(Model model, Map<String, String> result) {

        ModConfig modConfig = model.getModConfig();
        String userId = modConfig.getUserID();
        String userName = UserCache.instance().getNickName(Integer.valueOf(userId));

        Map<String, Object> dataMap = new HashMap<String, Object>();

        String extraReturnFields = Configuration.getInstance().getProperty(Constants.RETURN_FIELDS);
        if (StringUtils.isNotEmpty(extraReturnFields)) {
            String[] columns = extraReturnFields.split(",");
            for (String col : columns) {
                dataMap.put(col, result.get(col));
            }
        }

        String id = result.get("_id");
        String[] pari = id.split("_");
        dataMap.put("SCode", pari[0]);
        dataMap.put("MarketType", pari[1]);

        StrategyConfig strategyConfig = model.getStrategyConfig();
        if (strategyConfig != null) {
            dataMap.put("StrategyName", strategyConfig.getStrategyName());
            dataMap.put("StrategyID", strategyConfig.getStrategyID());
        }
        dataMap.put("UserID", userId);
        dataMap.put("UserName", userName);

        //add date
        Date date = new Date();
        dataMap.put("TDate", DateUtils.parseDate(date,DateUtils.TIME_FORMAT_D));
        dataMap.put("TTime", DateUtils.parseDate(date,DateUtils.TIME_FORMAT_II));
        kafkaProcessor.process(dataMap);
    }

    private boolean evalRule(Rule rule, Map<String, String> dataResult) {
        boolean flag = true;
        try {
            /**
             * 如果是用户自定义的规则需要先处理 自定义变量的值
             */
            Map<String, Object> ctx = new HashMap<String, Object>();
            ctx.putAll(dataResult);
            if (rule.isContainUserVar()) {
                List<UserVari> userVaris = rule.getUserVaris();

                for (UserVari userVari : userVaris) {
                    UserVariConfig userVariConfig = userVari.getVariConfig();
                    String userVarContext = userVariConfig.getContent();
                    Object result = ESModelQueryLoad.expression.eval(userVarContext, dataResult);
                    ctx.put(userVariConfig.getVariCode(), result);
                }
                flag = (Boolean) ESModelQueryLoad.expression.eval(rule.getParsedContext(), ctx);
            } else {
                String expr = rule.getParsedContext();
                flag = (Boolean) ESModelQueryLoad.expression.eval(expr, ctx);
            }
        } catch (Throwable e) {
//            logger.error("eval Rule error {} {}", dataResult, rule.getParsedContext(), e);
            flag = false;
        }
        return flag;
    }

    private void getAllColumns(List<Model> models, Set<String> needFields) {

        needFields.add("_id");

        for (Model model : models) {
            for (Rule rule : model.getRules()) {
                Set<String> extrVaris = rule.getExtraVariDeps();
                if (extrVaris != null) {
                    needFields.addAll(extrVaris);
                }
                List<Vari> varis = rule.getVars();
                if (varis != null) {
                    for (Vari vari : varis) {
                        needFields.add(vari.getConfig().getVariCode());
                    }
                }
            }
        }

        String extraReturnFields = Configuration.getInstance().getProperty(Constants.RETURN_FIELDS);
        if (StringUtils.isNotEmpty(extraReturnFields)) {
            needFields.addAll(Arrays.asList(extraReturnFields.split(",")));
        }
    }

    public void setKafkaProcessor(KafkaProcessor kafkaProcessor) {
        this.kafkaProcessor = kafkaProcessor;
    }

    public void setEsSqlClient(EsSqlClient esSqlClient) {
        this.esSqlClient = esSqlClient;
    }

    public void setExpression(Expression expression) {
        this.expression = expression;
    }
}
